package com.bw.ads;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import com.bw.bean.ProductStats;
import com.bw.util.ClickHouseUtil;
import com.bw.util.DateFormatUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;
import java.util.HashSet;

public class Test5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        env.setParallelism(1);

        DataStream<String> orderDetailStream = env.addSource(MyKafkaUtil.getKafkaConsumer("dws_trade_orders", "test11116"));


        SingleOutputStreamOperator<ProductStats> etlStream1 = orderDetailStream.flatMap(new FlatMapFunction<String, ProductStats>() {
            @Override
            public void flatMap(String s, Collector<ProductStats> collector) throws Exception {
                try {
                    if (s != null) {
                        JSONObject jsonObject = JSON.parseObject(s);
                        String sku_id = jsonObject.getString("product_sku_id");
                        Integer orderInfoId = jsonObject.getInteger("order_info_id");
                        if (sku_id != null) {
                            ProductStats productStats = JSON.parseObject(s, ProductStats.class);
                            HashSet<Integer> ids = new HashSet<>();
                            ids.add(orderInfoId);
                            productStats.setOrderIdSet(ids);
                            collector.collect(productStats);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        etlStream1.print();
//        // 添加水位线
        SingleOutputStreamOperator<ProductStats> orderInfoStream = etlStream1.assignTimestampsAndWatermarks(WatermarkStrategy.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(1)).withTimestampAssigner(new SerializableTimestampAssigner<ProductStats>() {
            @Override
            public long extractTimestamp(ProductStats ProductStats, long l) {
                return Long.parseLong(ProductStats.getCreate_time())/1000;
            }
        }));
//
        SingleOutputStreamOperator<ProductStats> reduceStream = orderInfoStream.keyBy(new KeySelector<ProductStats, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(ProductStats ProductStats) throws Exception {
                return Tuple2.of(ProductStats.getRegion_id() + "", ProductStats.getProduct_sku_id() + "");
            }
        }).window(SlidingEventTimeWindows.of(Time.days(1),Time.seconds(30))).reduce(new ReduceFunction<ProductStats>() {
            @Override
            public ProductStats reduce(ProductStats o1, ProductStats o2) throws Exception {
                ProductStats productStats = new ProductStats();
                productStats.setSku_num(o1.getSku_num() + o2.getSku_num());

                HashSet<Integer> ids = new HashSet<>();
                ids.addAll(o1.getOrderIdSet());
                ids.addAll(o2.getOrderIdSet());

                productStats.setOrderIdSet(ids);
                productStats.setAmount(o1.getAmount().add(o2.getAmount()));
                productStats.setOrderIdSet(o1.getOrderIdSet());
                productStats.setRegion_id(o1.getRegion_id());
                productStats.setRegion_name(o1.getRegion_name());
                productStats.setProduct_sku_id(o1.getProduct_sku_id());
                productStats.setProduct_spu_name(o1.getProduct_spu_name());
                System.out.println("productStats = " + productStats);
                return productStats;
            }
        }, new WindowFunction<ProductStats, ProductStats, Tuple2<String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple2<String, String> stringStringTuple2, TimeWindow timeWindow, Iterable<ProductStats> iterable, Collector<ProductStats> collector) throws Exception {
                String start = DateFormatUtil.tsToDateTime(timeWindow.getStart());
                String end = DateFormatUtil.tsToDateTime(timeWindow.getEnd());
                ProductStats productStats = iterable.iterator().next();
                productStats.setStt(start);
                productStats.setEdt(end);
                productStats.setOrder_ct((long) productStats.getOrderIdSet().size());
                productStats.setTs(System.currentTimeMillis());
                collector.collect(productStats);
            }
        });
        reduceStream.print();
        reduceStream.addSink(ClickHouseUtil.sink("insert into sku_yk8 (stt,edt,product_sku_id,product_spu_id,product_spu_name,sku_num,region_id,region_name,amount,create_time,order_ct,ts) values(?,?,?,?,?,?,?,?,?,?,?,?)"));
        env.execute();
    }
}
